package com.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author : ranzlupup
 * @date : 2023/3/4 17:10
 */
object TransformStreamWordCount {
    def main(args: Array[String]): Unit = {

        val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
        val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(3))

        val linesStream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 9999)
        val worAndCountDStream: DStream[(String, Int)] = linesStream.transform(rdd => {
            val words: RDD[String] = rdd.flatMap(_.split(" "))
            val wordAndOne: RDD[(String, Int)] = words.map((_, 1))
            val wordCount: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)
            wordCount
        })
        worAndCountDStream.print()

        ssc.start()
        ssc.awaitTermination()
    }
}
